package com.qingyunge.app.dws;

import com.qingyunge.app.func.SplitFunction;
import com.qingyunge.bean.DwsShiYinNumBean;
import com.qingyunge.bean.DwsShiYunTableBean;
import com.qingyunge.util.DateFormatUtil;
import com.qingyunge.util.MyClickHouseUtil;
import com.qingyunge.util.MyKafkaUtil;
import com.qingyunge.util.MyfenPriceUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Dws_shiyun_get_num_window {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String topic = "sku_eval_topic";
        String gourpid = "Dws_user_pingLun_time";
        // TODO 创建表
        tableEnv.executeSql("create table dws_shiyun_get_num_table( " +
                " `id` bigint, " +
                "    `skuId` string, " +
                "    `name` string, " +
                "    `price` double, " +
                "    `imgUrl` string, " +
                "    `evalType` string, " +
                "    `evalNum` bigint, " +
                "    `porView` int, " +
                "    `evaluate` string, " +
                "    `proconf` string, " +
                "    `date` string, " +
                "    `location` string, " +
                "    `ts` bigint " +
                "    )"+ MyKafkaUtil.getKafkaDDL(topic,gourpid));
        tableEnv.createTemporaryFunction("SplitFunction",SplitFunction.class);
        Table keyWordTable = tableEnv.sqlQuery("select  " +
                "    word AS keyword, " +
                "    price, " +
                "    `date`, " +
                "    location, " +
                "    skuId, " +
                "    ts " +
                "  from dws_shiyun_get_num_table, " +
                "LATERAL TABLE(SplitFunction(evaluate))");
        DataStream<DwsShiYunTableBean> dwsShiYunTableBeanDS = tableEnv.toAppendStream(keyWordTable, DwsShiYunTableBean.class);
//        dwsShiYunTableBeanDS.print("======");
        // TODO 关键词 年月  季度  价格区间 skuClass num
        SingleOutputStreamOperator<DwsShiYinNumBean> dwsShiYinNumBeanSOD = dwsShiYunTableBeanDS.flatMap(new FlatMapFunction<DwsShiYunTableBean, DwsShiYinNumBean>() {
            @Override
            public void flatMap(DwsShiYunTableBean dwsShiYunTableBean, Collector<DwsShiYinNumBean> collector) throws Exception {
                String keyword = dwsShiYunTableBean.getKeyword();
                String date = dwsShiYunTableBean.getDate();
                String time = null;
                String quarter = null;
                try {
                    time = date.substring(0,7);
                    int month = Integer.parseInt(date.substring(5, 7));
                    quarter = MyfenPriceUtil.getQuarter(month);
                } catch (Exception e) {
                    time = DateFormatUtil.toYmdHms(System.currentTimeMillis()).substring(0, 7);
                    int month = Integer.parseInt(time.substring(5, 7));
                    quarter = MyfenPriceUtil.getQuarter(month);
                }
                Double price = dwsShiYunTableBean.getPrice();
                String priceQj = MyfenPriceUtil.getSalaryRange(price);
                String skuClass = dwsShiYunTableBean.getSkuId().substring(0, 1);
                Long ts = dwsShiYunTableBean.getTs();
                DwsShiYinNumBean dwsShiYinNumBean = new DwsShiYinNumBean(keyword, time, quarter,
                        priceQj, skuClass, 1L,price,ts);
                collector.collect(dwsShiYinNumBean);
            }
        });
        SingleOutputStreamOperator<DwsShiYinNumBean> dwsShiyunWaterMark = dwsShiYinNumBeanSOD.assignTimestampsAndWatermarks(WatermarkStrategy.<DwsShiYinNumBean>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<DwsShiYinNumBean>() {
            @Override
            public long extractTimestamp(DwsShiYinNumBean dwsShiYinNumBean, long l) {
                return dwsShiYinNumBean.getTs();
            }
        }));
        // TODO  分组 开窗 聚合
        SingleOutputStreamOperator<DwsShiYinNumBean> reduceShiYunDS = dwsShiyunWaterMark.keyBy(new KeySelector<DwsShiYinNumBean, Tuple5<String, String, String,String, String>>() {
            @Override
            // TODO 关键词 年月 季度  价格区间  skuClass
            public Tuple5<String,String, String, String, String> getKey(DwsShiYinNumBean value) throws Exception {
                return Tuple5.of(value.getKeyword(), value.getTime(), value.getQuarter(), value.getPriceQj(), value.getSkuClass());
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<DwsShiYinNumBean>() {
                    @Override
                    public DwsShiYinNumBean reduce(DwsShiYinNumBean dwsShiYinNumBean, DwsShiYinNumBean t1) throws Exception {
                        dwsShiYinNumBean.setNum(dwsShiYinNumBean.getNum() + t1.getNum());
                        Double pricet = Math.round((dwsShiYinNumBean.getPriceSum() + t1.getPriceSum()) * 100.0) / 100.0;
                        dwsShiYinNumBean.setPriceSum(pricet);
                        return dwsShiYinNumBean;
                    }
                });
        reduceShiYunDS.print("=========>");
        // TODO  写入clickhouse
        reduceShiYunDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_shiyin_count_table values (?,?,?,?,?,?,?,?)"));
        //TODO 启动任务
        env.execute();
    }
}
